package cn.jbooter.mq.autoconfigure.rocketmq.coms;

import java.util.List;
import java.util.UUID;
import javax.annotation.PreDestroy;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.Assert;

import cn.jbooter.coms.exceptions.FrameworkException;
import cn.jbooter.coms.utils.StringUtil;
import cn.jbooter.mq.autoconfigure.rocketmq.enu.RocketConsumeMode;
import cn.jbooter.mq.autoconfigure.rocketmq.properties.RocketConsumerInitInfo;
import cn.jbooter.mq.autoconfigure.rocketmq.properties.RocketProperties;

/**
 * push comsumer抽象类,业务consumer需要继承此抽象类型
 * 注:在消费的过程中为线程非安全
 * @author HeJian
 *
 */
public abstract class AbstractRocketPushConsumer implements IRocketConsumer {
	private static final Logger logger = LoggerFactory.getLogger(AbstractRocketPushConsumer.class);
	
	/** rocketmq配置信息 **/
	@Autowired(required=false)
	private RocketProperties rocketProperties;
	/**consumer**/
	private DefaultMQPushConsumer consumer;
	/** consumer初始化信息 **/
	private RocketConsumerInitInfo consumerInitInfo;
	
	public RocketProperties getRocketProperties() {
		return rocketProperties;
	}
	public void setRocketProperties(RocketProperties rocketProperties) {
		this.rocketProperties = rocketProperties;
	}
	public DefaultMQPushConsumer getConsumer() {
		return consumer;
	}
	public void setConsumer(DefaultMQPushConsumer consumer) {
		this.consumer = consumer;
	}
	public RocketConsumerInitInfo getConsumerInitInfo() {
		return consumerInitInfo;
	}
	public void setConsumerInitInfo(RocketConsumerInitInfo consumerInitInfo) {
		this.consumerInitInfo = consumerInitInfo;
	}
	
	/**
	 * push comsumer 初始化
	 */
	@Override
	public void init(RocketConsumerInitInfo init) throws Exception{
		//参数检验
		if(this.rocketProperties==null) {
			throw new FrameworkException("rocketmq配置信息为空!");
		}
		if(init==null) {
			throw new FrameworkException("consumer 初始化参数为空!");
		}
		this.consumerInitInfo = init;
		if(StringUtil.isEmpty(init.getNamesrvAddr())) {
			init.setNamesrvAddr(this.rocketProperties.getNamesrvAddr());
		}
		Assert.hasText(init.getNamesrvAddr(), "Name Server Address未配置! bean : " + init.getBeanName());
		Assert.hasText(init.getConsumerGroup(), "consumerGroup未定义! bean : " + init.getBeanName());
		Assert.hasText(init.getTopic(), "topic未定义! bean : " + init.getBeanName());
		Assert.notNull(init.getMessageModel(), "messageModel未定义! bean : " + init.getBeanName());
		Assert.notNull(init.getConsumeMode(), "consumeMode未定义! bean : " + init.getBeanName());
		Assert.notEmpty(init.getTag(), "tag未定义! bean : " + init.getBeanName());
		
        // 初始化push consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(init.getConsumerGroup());
        consumer.setNamesrvAddr(init.getNamesrvAddr());
        consumer.setMessageModel(init.getMessageModel());
        consumer.subscribe(init.getTopic(), StringUtil.join(init.getTag(), "||"));
        consumer.setInstanceName(UUID.randomUUID().toString());
        consumer.setVipChannelEnabled(this.rocketProperties.isVipChannelEnabled());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        if (RocketConsumeMode.CONCURRENTLY.equals(init.getConsumeMode())) {
        	//无序消费
        	consumer.registerMessageListener(new MessageListenerConcurrently() {
				@Override
				public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
					return consumeConcurrentlyMessage(msgs,context);
				}
        	});
        	
        } else if (RocketConsumeMode.ORDERLY.equals(init.getConsumeMode())) {
        	//顺序消费
        	consumer.registerMessageListener(new MessageListenerOrderly() {
				@Override
				public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
					return consumeOrderlyMessage(msgs,context);
				}
        	});
        } else {
            throw new FrameworkException("未知的consume mode ! 仅支持CONCURRENTLY和ORDERLY ! bean:"+init.getBeanName());
        }
        consumer.start();
        this.consumer = consumer;
        logger.info("bean:{} the consumer is started !",init.getBeanName());
	}

	/**
	 * push consumer的销毁
	 */
	@Override
	@PreDestroy
	public void destroy() throws Exception{
		if(this.consumer!=null) {
			logger.info("shutdown the consumer : {}",this.consumerInitInfo.getBeanName());
			this.consumer.shutdown();
		}
	}
	
	/**
	 * 无序消息的消费
	 * 
	 * 消费失败会重新消费
	 * @param msgs
	 * @param Context
	 * @return
	 */
	public ConsumeConcurrentlyStatus consumeConcurrentlyMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		for(MessageExt messageExt : msgs) {
			logger.info("receive msgId: {}, topic : {}, tags : {}" , messageExt.getMsgId(), messageExt.getTopic(),messageExt.getTags());
            if( !process(messageExt)) {
            	logger.warn("consume fail , ask for re-consume , msgId: {}, topic : {}, tags : {}" , messageExt.getMsgId(), messageExt.getTopic(),messageExt.getTags());
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return  ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}
	
	/**
	 * 有序消息的消费
	 * @param msgs
	 * @param Context
	 * @return
	 */
	public ConsumeOrderlyStatus  consumeOrderlyMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
		for(MessageExt messageExt : msgs) {
            logger.info("receive msgId: {}, topic : {}, tags : {}" , messageExt.getMsgId(), messageExt.getTopic(),messageExt.getTags());
            if( !process(messageExt)) {
            	logger.warn("consume fail , ask for re-consume , msgId: {}, topic : {}, tags : {}" , messageExt.getMsgId(), messageExt.getTopic(),messageExt.getTags());
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        }
        return  ConsumeOrderlyStatus.SUCCESS;
	}

}
